package com.sfx.mq.demo;

import com.sfx.mq.constants.MQConstant;
import com.sfx.mq.constants.TopicProperties;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.junit.Test;

import java.util.List;
import java.util.UUID;

/**
 * Created with IntelliJ IDEA.
 * Description:
 * User: sfx
 * Date: 2023-10-28
 * Time: 9:50
 */
public class KeysTest {

    @Test
    public void producer() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("producer-group");
        producer.setNamesrvAddr(MQConstant.NAME_SRV);
        producer.start();
        Message message = new Message();
        message.setTopic(TopicProperties.KEYS_TOPIC);
        String keys = UUID.randomUUID().toString();
        message.setKeys(keys);
        message.setBody("keys消息".getBytes());
        SendResult send = producer.send(message);
        System.out.println("发送消息成功 status is " + send.getSendStatus());
        System.out.println("发送消息唯一的业务标识(keys)" + keys);
        producer.shutdown();
    }


    @Test
    public void consumer() throws Exception {

        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer-group");
        consumer.setNamesrvAddr(MQConstant.NAME_SRV);
        consumer.subscribe(TopicProperties.KEYS_TOPIC,"*");

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                System.out.println("我是消费者 threadId is" + Thread.currentThread().getId());
                MessageExt messageExt = list.get(0);
                System.out.println("消费消息为 :" + new String(messageExt.getBody()));
                System.out.println("消费消息的唯一业务标识为: " + messageExt.getKeys());
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.start();
        // 挂起JVM
        System.in.read();
        consumer.shutdown();
    }
}